# Define your item pipelines here
#
# Don't forget to add your pipeline to the ITEM_PIPELINES setting
# See: https://docs.scrapy.org/en/latest/topics/item-pipeline.html


# useful for handling different item types with a single interface
import json

from itemadapter import ItemAdapter

from kafka import KafkaProducer
from SpiderDemo9.items import WeiBoArticleItem, UserItem, CommentItem


class SpiderDemo9Pipeline:
    # 创建一个数据库连接
    def __init__(self):
        self.producer = KafkaProducer(bootstrap_servers="master:9092,node1:9092,node2:9092")

    def process_item(self, item, spider):
        # # 使用数据库连接，将item中包含的数据写入外部系统
        itemValue = json.dumps(dict(item), ensure_ascii=False).encode("utf8")
        topic = ""

        # item 有多种类型，不同的Item类型需要发送到不同的Kafka Topic
        if isinstance(item, WeiBoArticleItem):
            topic = "ods_kafka_weibo"
        elif isinstance(item, UserItem):
            topic = "ods_kafka_user"
        elif isinstance(item, CommentItem):
            topic = "ods_kafka_comment"
        else:
            topic = "errorTopic"
        self.producer.send(topic, value=itemValue)
        return item

    def close_spider(self, spider):
        self.producer.close()
